package com.practice.gmall.realtime.app.dwd;

import com.practice.gmall.realtime.app.BaseSQLApp;
import com.practice.gmall.realtime.util.MyKafkaUtil;
import com.practice.gmall.realtime.util.MySQLUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * ClassName: DWD_03_TradeOrderDetail
 * Package: com.practice.gmall.realtime.app.dwd
 * Description: 主要任务 将MySQL中order_detail表的数据进行维表关联，再将关联好后的表写出到Kafka dwd_order_detail主题中
 * <p>
 * 下单事务事实表
 *     粒度: 最小粒度, 商品粒度
 *     维度:  省份 sku spu tm c3 c2 c1
 *     度量:  原始总金额 分摊总金额  优惠减免的金额 活动减免
 *
 * @Author lzy.ethan@qq.com
 * @Create 2023-02-14 14:24
 */
public class DWD_03_TradeOrderDetail extends BaseSQLApp {
    public static void main(String[] args) {
        new DWD_03_TradeOrderDetail().init(3337,"DWD_03_TradeOrderDetail",2);
    }

    @Override
    public void handle(StreamExecutionEnvironment env, StreamTableEnvironment tEnv) {
        // 1.  read ods_db
        MyKafkaUtil.readTopicOdsDb(tEnv,"DWD_03_TradeOrderDetail");
        // 2. filter order_detail data
        Table order_detail = tEnv.sqlQuery("" +
                "SELECT   " +
                "    `data`['id'] id,  " +
                "    `data`['order_id'] order_id,  " +
                "    `data`['sku_id'] sku_id,  " +
                "    `data`['sku_name'] sku_name,  " +
                "    `data`['order_price'] order_price,  " +
                "    `data`['sku_num'] sku_num,  " +
                "    `data`['create_time'] create_time,  " +
                "    `data`['source_type'] source_type,  " +
                "    `data`['source_id'] source_id,  " +
                "    cast(cast(data['sku_num'] as decimal(16,2)) * cast(data['order_price'] as decimal(16,2)) as  string) split_original_amount, " +
                "    `data`['split_total_amount'] split_total_amount,  " +
                "    `data`['split_activity_amount'] split_activity_amount,  " +
                "    `data`['split_coupon_amount'] split_coupon_amount,  " +
                "    `ts` ,  " +
                "    `pt` " +
                "FROM ods_db  " +
                "WHERE   " +
                "    `database` = 'gmall2022'   " +
                "    AND `table` = 'order_detail'   " +
                "    AND `type` =  'insert'  ");

        tEnv.createTemporaryView("order_detail",order_detail);

        //tEnv.toAppendStream(order_detail, Row.class).print(">>>>>>>>>");// test print out

        // 3.0 load dimension table base_dic
        tEnv.executeSql(MySQLUtil.getBaseDicLookUp());

        // 3.1 load dimension table order_info
        Table order_info = tEnv.sqlQuery("" +
                "SELECT  " +
                "    `data`['id'] id, " +
                "    `data`['user_id'] user_id, " +
                "    `data`['total_amount'] total_amount, " +
                "    `data`['create_time'] create_time, " +
                "    `data`['operate_time'] operate_time, " +
                "    `data`['expire_time'] expire_time, " +
                "    `data`['province_id'] province_id, " +
                "    `data`['activity_reduce_amount'] activity_reduce_amount, " +
                "    `data`['coupon_reduce_amount'] coupon_reduce_amount, " +
                "    `data`['original_total_amount'] original_total_amount " +
                "FROM ods_db " +
                "WHERE  " +
                "    `database` = 'gmall2022'  " +
                "    AND `table` = 'order_info'  " +
                "    AND `type` =  'insert'");

        tEnv.createTemporaryView("order_info",order_info);

        // 3.2 load dimension table order_detail_coupon
        Table order_detail_coupon = tEnv.sqlQuery("" +
                "SELECT   " +
                "    data['order_detail_id'] order_detail_id,  " +
                "    data['coupon_id'] coupon_id   " +
                "FROM ods_db   " +
                "WHERE  " +
                "    `database`='gmall2022'   " +
                "    AND `table`='order_detail_coupon'   " +
                "    AND `type`='insert'");

        tEnv.createTemporaryView("order_detail_coupon",order_detail_coupon);

        // 3.3 load dimension table order_detail_activity
        Table order_detail_activity = tEnv.sqlQuery("" +
                "SELECT   " +
                "    `data`['order_detail_id'] order_detail_id,  " +
                "    `data`['activity_id'] activity_id,  " +
                "    `data`['activity_rule_id'] activity_rule_id   " +
                "FROM ods_db   " +
                "WHERE  " +
                "    `database`='gmall2022'   " +
                "    AND `table`='order_detail_activity'   " +
                "    AND `type`='insert'");

        tEnv.createTemporaryView("order_detail_activity",order_detail_activity);

        // 4. execute join of these tables
        Table result = tEnv.sqlQuery("" +
                "SELECT  " +
                "    od.id   id,    " +
                "    od.order_id   order_id,    " +
                "    oi.user_id   user_id,   " +
                "    od.sku_id   sku_id,    " +
                "    od.sku_name   sku_name,    " +
                "    od.order_price   order_price,    " +
                "    od.sku_num   sku_num,    " +
                "    date_format(od.create_time, 'yyyy-MM-dd')   date_id,    " +
                "    oi.province_id  province_id, " +
                "    od.source_type   source_type_code, " +
                "    dic.dic_name    source_type_name,    " +
                "    od.source_id   source_id, " +
                "    oda.activity_id as activity_id, " +
                "    oda.activity_rule_id , " +
                "    odc.coupon_id as coupon_id, " +
                "    od.split_original_amount,    " +
                "    od.split_total_amount , " +
                "    od.split_activity_amount , " +
                "    od.split_coupon_amount , " +
                "    `ts` " +
                "FROM " +
                "    order_detail od JOIN order_info oi ON od.order_id = oi.id " +
                "    LEFT JOIN order_detail_activity oda ON od.id = oda.order_detail_id " +
                "    LEFT JOIN order_detail_coupon odc ON od.id = odc.order_detail_id " +
                "    JOIN base_dic  FOR SYSTEM_TIME AS OF od.pt as dic " +
                "        ON od.source_type = dic.dic_code");

        //tEnv.toRetractStream(result, Row.class).print(">>>>>");// test print

        // 5. create output table,i.e. execute dwd_order_detail DDL, for update stream, use upsert kafka connector
        // todo 什么时候使用upsertKafka
        tEnv.executeSql("" +
                "CREATE TABLE dwd_order_detail( " +
                "    id  string,    " +
                "    order_id  string,    " +
                "    user_id  string,   " +
                "    sku_id  string,    " +
                "    sku_name  string,    " +
                "    order_price  string,    " +
                "    sku_num  string,    " +
                "    date_id  string,    " +
                "    province_id  string, " +
                "    source_type_code  string, " +
                "    source_type_name  string, " +
                "    source_id  string, " +
                "    activity_id  string, " +
                "    activity_rule_id   string, " +
                "    coupon_id  string, " +
                "    split_original_amount  string,    " +
                "    split_total_amount   string, " +
                "    split_activity_amount   string, " +
                "    split_coupon_amount   string, " +
                "    `ts` BIGINT, " +
                "    primary key(id) not enforced " +
                ")"  + MyKafkaUtil.getKafkaUpsertSinkDDL("dwd_order_detail"));
        
        // 6. execute write to kafka
        result.executeInsert("dwd_order_detail");


        /*try {
            env.execute();
        } catch (Exception e) {
            throw new RuntimeException(e);
        }*/
    }
}
